package com.duoduo.mcc.service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.duoduo.mcc.bean.MccMessage;
import com.duoduo.mcc.thread.MccProcessThread;

/**
 * 消息接收入口
 * @author cgs1999@126.com
 * @date 2018年11月1日下午6:13:58
 */
public class MccReceiveService implements Runnable {
	private Logger logger = LoggerFactory.getLogger(getClass());
	private MccCollectionService mccCollectionService;
	private MccQueueService mccQueueService;
	private static ConcurrentLinkedQueue<MccMessage> messageQueue = new ConcurrentLinkedQueue<MccMessage>();

	public MccReceiveService(MccCollectionService mccCollectionService, MccQueueService mccQueueService) {
		this.mccCollectionService = mccCollectionService;
		this.mccQueueService = mccQueueService;
	}
	
	public synchronized void addMessage(MccMessage message) {
		if(messageQueue==null) {
			if (logger.isDebugEnabled()) {
				logger.debug("当前接收线程没有队列，需要创建");
			}
			messageQueue = new ConcurrentLinkedQueue<MccMessage>();
		}
		messageQueue.offer(message);
		if (logger.isDebugEnabled()) {
			logger.debug("已经添加消息 {} 到queue中", message.toString());
		}
	}

	public void run() {
		while(true) {
			List<MccMessage> messageList = new ArrayList<MccMessage> (0);
			while(!messageQueue.isEmpty()) {
				for(int i=0; i<100; i++) {
					MccMessage message = messageQueue.poll();
					if(message!=null) {
						messageList.add(message);
					}
				}
				
				for(MccMessage message : messageList) {
					processMsg(message);
				}
			}
			
			try {
				logger.debug("消息处理完成，本次共处理{}条消息，线程等待1秒...", messageList.size());
				Thread.sleep(1000);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	private void processMsg(MccMessage message) {
		// 判断操作的数据是否有过处理
		message.setTimeMillis(System.currentTimeMillis());
		mccQueueService.addMsg(message.getE164(), message);

		if (!mccCollectionService.conSet(message.getE164())) {
			if (logger.isDebugEnabled()) {
				logger.debug("队列中没有该字段的消息，需要启动线程来处理，消息 {}", message.getOperaData());
			}
			MccProcessThread processThread = new MccProcessThread(mccCollectionService, mccQueueService);
			processThread.setOrdered(1);
			processThread.setE164(message.getE164());
			mccCollectionService.addSet(message.getE164());

			try {
				mccQueueService.setRestartTime(message.getE164(), System.currentTimeMillis());// 每次启动新线程更新时间戳
				Thread thread = new Thread(processThread);
				thread.run();
			} catch (Exception e) {
				logger.error("多线程处理出错，该消息{}直接使用普通方法处理, 错误消息{}", message.getOperaData(), e.getMessage());

			}
		} else {
			if (logger.isDebugEnabled()) {
				logger.debug("队列中已经该字段的消息，判断线程处理是否超时，消息 {}", message.getOperaData());
				logger.debug("队列中会议号为：" + message.getE164() + "的线程时间戳为："
						+ mccQueueService.getRestartTime(message.getE164()));
			}
			if (!(mccQueueService.getRestartTime(message.getE164()) == null || (System.currentTimeMillis()
					- mccQueueService.getRestartTime(message.getE164()) < 60000))) {// 1分钟内未处理完1条消息
				logger.error("1分钟内未处理对应e164={}队列中的第一条消息，默认重新启动一个新线程", message.getE164());
				MccProcessThread processThread = new MccProcessThread(mccCollectionService, mccQueueService);
				processThread.setOrdered(1);
				processThread.setE164(message.getE164());

				try {
					mccQueueService.setRestartTime(message.getE164(), System.currentTimeMillis());// 每次启动新线程更新时间戳
					Thread thread = new Thread(processThread);
					thread.run();
				} catch (Exception e) {
					logger.error("多线程处理出错，该消息{}直接使用普通方法处理, 错误消息{}", null, e.getMessage());

				}
			}
		}
	}
}
